package com.alibaba.otter.canal.connector.kafka.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.connector.core.config.CanalConstants;
import com.alibaba.otter.canal.connector.core.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.consumer.OggMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer;
import com.alibaba.otter.canal.connector.core.spi.SPI;
import com.alibaba.otter.canal.connector.core.util.MessageUtil;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * Kafka consumer SPI 实现
 *
 * @author rewerma @ 2020-02-01
 * @version 1.0.0
 */
@SPI("kafka")
public class CanalKafkaConsumer implements CanalMsgConsumer {

    private static final String      PREFIX_KAFKA_CONFIG = "kafka.";

    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private static SimpleDateFormat sdf2 = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");

    private static JSONObject messageFormats = JSON.parseObject("{\"canal\":[\"data\", \"database\", \"es\", \"id\", \"isDdl\", \"mysqlType\", \"old\", \"pkNames\", \"sql\", \"sqlType\", \"table\", \"ts\", \"type\"],\"ogg\":[\"table\", \"op_type\", \"op_ts\", \"current_ts\", \"pos\", \"after\", \"before\"]}");

//    private static List<String> canalJsonKeys = Arrays.asList(new String[]{"data", "database", "es", "id", "isDdl", "mysqlType", "old", "pkNames", "sql", "sqlType", "table", "ts", "type"});
//    private static List<String> oggJsonKeys = Arrays.asList(new String[]{"table", "op_type", "op_ts", "current_ts", "pos", "after", "before"});

    private KafkaConsumer<String, ?> kafkaConsumer;
    private boolean                  flatMessage         = true;
    private String                   topic;

    private Map<Integer, Long>       currentOffsets      = new ConcurrentHashMap<>();
    private Properties               kafkaProperties     = new Properties();

    @Override
    public void init(Properties properties, String topic, String groupId) {
        this.topic = topic;

        Boolean flatMessage = (Boolean) properties.get(CanalConstants.CANAL_MQ_FLAT_MESSAGE);
        if (flatMessage != null) {
            this.flatMessage = flatMessage;
        }
        for (Map.Entry<Object, Object> entry : properties.entrySet()) {
            String k = (String) entry.getKey();
            Object v = entry.getValue();
            if (k.startsWith(PREFIX_KAFKA_CONFIG) && v != null) {
                kafkaProperties.put(k.substring(PREFIX_KAFKA_CONFIG.length()), v);
            }
        }
        kafkaProperties.put("group.id", groupId);
        kafkaProperties.put("key.deserializer", StringDeserializer.class);
        kafkaProperties.put("client.id", UUID.randomUUID().toString().substring(0, 6));
    }

    @Override
    public void connect() {
        if (this.flatMessage) {
            kafkaProperties.put("value.deserializer", StringDeserializer.class);
            this.kafkaConsumer = new KafkaConsumer<String, String>(kafkaProperties);
        } else {
            kafkaProperties.put("value.deserializer", KafkaMessageDeserializer.class);
            this.kafkaConsumer = new KafkaConsumer<String, Message>(kafkaProperties);
        }
        kafkaConsumer.subscribe(Collections.singletonList(topic));
    }

    @SuppressWarnings("unchecked")
    @Override
    public List<CommonMessage> getMessage(Long timeout, TimeUnit unit) {
        if (!flatMessage) {
            ConsumerRecords<String, Message> records = (ConsumerRecords<String, Message>) kafkaConsumer.poll(unit.toMillis(timeout));
            if (!records.isEmpty()) {
                currentOffsets.clear();
                List<CommonMessage> messages = new ArrayList<>();
                for (ConsumerRecord<String, Message> record : records) {
                    if (currentOffsets.get(record.partition()) == null) {
                        currentOffsets.put(record.partition(), record.offset());
                    }
                    messages.addAll(MessageUtil.convert(record.value()));
                }
                return messages;
            }
        } else {
            ConsumerRecords<String, String> records = (ConsumerRecords<String, String>) kafkaConsumer.poll(unit.toMillis(timeout));

            if (!records.isEmpty()) {
                List<CommonMessage> messages = new ArrayList<>();
                currentOffsets.clear();
                for (ConsumerRecord<String, String> record : records) {
                    if (currentOffsets.get(record.partition()) == null) {
                        currentOffsets.put(record.partition(), record.offset());
                    }
                    CommonMessage flatMessages = convertMessage(record.value());
                    if (null != flatMessages) {
                        messages.add(flatMessages);
                    }
                }
                return messages;
            }
        }
        return null;
    }

    @Override
    public void rollback() {
        // 回滚所有分区
        if (kafkaConsumer != null) {
            for (Map.Entry<Integer, Long> entry : currentOffsets.entrySet()) {
                kafkaConsumer.seek(new TopicPartition(topic, entry.getKey()), currentOffsets.get(entry.getKey()));
                kafkaConsumer.commitSync();
            }
        }
    }

    @Override
    public void ack() {
        if (kafkaConsumer != null) {
            kafkaConsumer.commitSync();
        }
    }

    @Override
    public void disconnect() {
        if (kafkaConsumer != null) {
            kafkaConsumer.unsubscribe();
        }
        if (kafkaConsumer != null) {
            kafkaConsumer.close();
            kafkaConsumer = null;
        }
    }

    private CommonMessage convertMessage(String flatMessageJson) {
        CommonMessage commonMessage = null;
        try {
            JSONObject jsonObject = JSON.parseObject(flatMessageJson);
            Set<String> keys;
            JSONArray value;
            String format = "canal";
            for (Map.Entry<String, Object> kv : messageFormats.entrySet()) {
                value = (JSONArray) kv.getValue();
                keys = ((JSONObject) jsonObject.clone()).keySet();
                keys.removeAll(value);
                if (keys.size() == 0) {
                    format = kv.getKey();
                    break;
                }
            }
            if ("canal".equals(format)) {
                commonMessage = JSON.parseObject(flatMessageJson, CommonMessage.class);
            } else if ("ogg".equals(format)) {
                // 将ogg生产的消息格式转换为canal json格式
                OggMessage oggMessage = JSON.parseObject(flatMessageJson, OggMessage.class);
                commonMessage = new CommonMessage();
                commonMessage.setIsDdl(false);
                commonMessage.setData(oggMessage.getAfter());
                String table = oggMessage.getTable();
                commonMessage.setTable(table);
                if (table.contains(".")) {
                    String[] split = table.split("\\.");
                    if (split.length == 2) {
                        commonMessage.setDatabase(split[0]);
                        commonMessage.setTable(split[1]);
                    }
                }

                commonMessage.setOld(oggMessage.getBefore());

                switch (oggMessage.getOp_type()) {
                    case "I":
                        commonMessage.setType("INSERT");
                        break;
                    case "U":
                        commonMessage.setType("UPDATE");
                        break;
                    case "D":
                        commonMessage.setType("DELETE");
                        commonMessage.setOld(null);
                        commonMessage.setData(oggMessage.getBefore());
                        break;
                    case "T":
                        commonMessage.setType("TRUNCATE");
                        break;
                }

                commonMessage.setEs(formatDateStr(oggMessage.getOp_ts()));
                commonMessage.setTs(formatDateStr(oggMessage.getCurrent_ts()));
            }

        } catch (Exception e) {

        }
        return commonMessage;
    }

    private Long formatDateStr(String datetime) {
        Date date = null;
        try {
            date = sdf.parse(datetime);
        } catch (ParseException e) {
            try {
                date = sdf2.parse(datetime);
            } catch (ParseException ex) {

            }
        }
        return null == date ? new Date().getTime() : date.getTime();
    }

}
